package cn.doitedu.rtmk.demo4;

import cn.doitedu.rtmk.beans.UserEvent;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/22
 * @Desc: 学大数据，上多易教育
  **/
public class Demo4 {

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);


        // 创建kafka 行为明细topic的source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("dwd-events")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("doit43-8")
                .setClientIdPrefix("doitedu-c")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> streamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");
        SingleOutputStreamOperator<UserEvent> eventStream = streamSource.map(json -> JSON.parseObject(json, UserEvent.class));

        SingleOutputStreamOperator<String> hitMessages =
                eventStream
                        .keyBy(UserEvent::getUser_id)
                        .process(new KeyedProcessFunction<Long, UserEvent, String>() {

                            HashMap<String,RuleCalculator> calculatorPool = new HashMap<>();

                            @Override
                            public void open(Configuration parameters) throws Exception {

                                // 构造规则001的运算机对象
                                String rule001ParamJson = "{\n" +
                                        "  \"rule_id\": \"rule-A-001\",\n" +
                                        "  \"dynamic_profile_condition1\": {\n" +
                                        "    \"event_id\": \"R\",\n" +
                                        "    \"operator\": \">\",\n" +
                                        "    \"count\": 6\n" +
                                        "  },\n" +
                                        "  \"dynamic_profile_condition2\": {\n" +
                                        "    \"event_id\": \"K\",\n" +
                                        "    \"property_name\": \"p2\",\n" +
                                        "    \"operator\": \">\",\n" +
                                        "    \"max_min_bs\": 8\n" +
                                        "  },\n" +
                                        "  \"trigger_condition\": {\n" +
                                        "    \"event_id\": \"M\",\n" +
                                        "    \"property_name\": \"p1\",\n" +
                                        "    \"operator\": \">\",\n" +
                                        "    \"compare_value\": 2\n" +
                                        "  }\n" +
                                        "}";

                                // 从模型运算机，构建一个具体规则的运算机对象
                                RuleModel_A_Calculator rule001Calculator = new RuleModel_A_Calculator();
                                rule001Calculator.init(getRuntimeContext(),rule001ParamJson);
                                // 将初始化好的规则运算机对象，放入运算机池
                                calculatorPool.put("rule-A-001",rule001Calculator);




                                String rule002ParamJson = "{\n" +
                                        "  \"rule_id\": \"rule-A-002\",\n" +
                                        "  \"dynamic_profile_condition1\": {\n" +
                                        "    \"event_id\": \"V\",\n" +
                                        "    \"operator\": \">\",\n" +
                                        "    \"count\": 2\n" +
                                        "  },\n" +
                                        "  \"dynamic_profile_condition2\": {\n" +
                                        "    \"event_id\": \"N\",\n" +
                                        "    \"property_name\": \"p1\",\n" +
                                        "    \"operator\": \">\",\n" +
                                        "    \"max_min_bs\": 3\n" +
                                        "  },\n" +
                                        "  \"trigger_condition\": {\n" +
                                        "    \"event_id\": \"S\",\n" +
                                        "    \"property_name\": \"p1\",\n" +
                                        "    \"operator\": \"=\",\n" +
                                        "    \"compare_value\": 3\n" +
                                        "  }\n" +
                                        "}";
                                RuleModel_A_Calculator rule002Calculator = new RuleModel_A_Calculator();
                                rule002Calculator.init(getRuntimeContext(),rule002ParamJson);
                                // 将初始化好的规则运算机对象，放入运算机池
                                calculatorPool.put("rule-A-002",rule002Calculator);

                            }

                            @Override
                            public void processElement(UserEvent userEvent, KeyedProcessFunction<Long, UserEvent, String>.Context ctx, Collector<String> out) throws Exception {

                                // 调规则运算机对象的 calculate()方法
                                for (Map.Entry<String, RuleCalculator> entry : calculatorPool.entrySet()) {
                                    RuleCalculator calculator = entry.getValue();
                                    calculator.calculate(userEvent,out);
                                }

                            }
                        });


        hitMessages.print();

        env.execute();




    }

}
